package com.demo.rocketmq.api;

import cn.hutool.core.date.DateUtil;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Description
 * @Author likz
 * @Date 2024/9/10 18:19
 **/
@Slf4j
@RestController
@RequestMapping("api/test/info")
@RequiredArgsConstructor
@Api(tags = "测试API-消息消息确认")
public class TestInfoAffirmController {

    private final RocketMQTemplate rocketMQTemplate;

    public static final String TOPIC = "infoConfirm";
    public static final String TOPIC_CANAL = "canal";

    @ApiOperation("消息确认")
    @PostMapping("/infoConfirm")
    public String infoConfirm() {
        String msgContent = DateUtil.now();
        log.info("[生产者] 发送消息：{}", msgContent);
        this.rocketMQTemplate.convertAndSend(TOPIC, msgContent);
        return "OK";
    }


    @ApiOperation("消息取消")
    @PostMapping("/canalInfo")
    public String canalInfo() {
        String msgContent = "我是一个取消的消息";
        Message msg = new Message(TOPIC, "tag", "messageKey", ("version=" + 1 + ", content=" + "content").getBytes());
        this.rocketMQTemplate.convertAndSend(TOPIC, msg);


        return "OK";
    }


    @Slf4j
    @Service
    @RocketMQMessageListener(
            topic = TOPIC, // 主题，指消费者组订阅的消息服务
            selectorExpression = "tag1",
            consumerGroup = "canal-consumer", // 消费者组，一个组可以有多个消费者，主要的作用是集群模式负载均衡的实现，广播模式的通知的实现
            consumeMode = ConsumeMode.CONCURRENTLY, // 控制消费模式，你可以选择并发或有序接收消息
            messageModel = MessageModel.CLUSTERING // 控制消息模式，广播模式：所有消费者都能接收到信息，集群模式：无论有多少个消费者，只有一个消费者能够接收到信息，也就是说消息一旦被消费了，其它消费者就不能消费该条消息

    )
    static class canalListener2 implements RocketMQListener {


        @Override
        public void onMessage(Object message) {

        }
    }







    @Slf4j
    @Service
    @RocketMQMessageListener(
            topic = TOPIC, // 主题，指消费者组订阅的消息服务
            consumerGroup = "infoConfirm-consumer", // 消费者组，一个组可以有多个消费者，主要的作用是集群模式负载均衡的实现，广播模式的通知的实现
            consumeMode = ConsumeMode.CONCURRENTLY, // 控制消费模式，你可以选择并发或有序接收消息
            messageModel = MessageModel.CLUSTERING // 控制消息模式，广播模式：所有消费者都能接收到信息，集群模式：无论有多少个消费者，只有一个消费者能够接收到信息，也就是说消息一旦被消费了，其它消费者就不能消费该条消息

    )
    static class infoConfirmListener2 implements RocketMQReplyListener {


        @Override
        public Object onMessage(Object message) {
            return null;
        }
    }


}
